package com.crazymaker.springcloud.message.service.impl;

import com.crazymaker.springcloud.common.util.ThreadUtil;
import com.crazymaker.springcloud.distribute.zookeeper.ZKClient;
import com.crazymaker.springcloud.message.config.ShardingRuleConfig;
import com.crazymaker.springcloud.message.core.LocalRegistryCenter;
import com.crazymaker.springcloud.standard.context.SpringContextUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.shardingsphere.orchestration.reg.listener.DataChangedEvent;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.UnsupportedEncodingException;
import java.util.concurrent.TimeUnit;

import static com.crazymaker.springcloud.message.config.ShardingRuleConfig.DEFAULT_SHARDING_RULE_CONFIG;

/**
 *
 */
@Slf4j
@Component
public class ShardingService implements ApplicationContextAware {

    public static final String SHARDING_CHANG_PATH = "/push-center/sharding_chang/config";

    @Resource
    ShardingRuleConfig shardingRuleConfig;

    /**
     * 替换sharding里的分表规则ActualDataNodes的值
     */
    public void replaceDataSource(String config) {

        CuratorFramework client = ZKClient.getSingleton().getClient();

        // 变更节点数据
        try {
            client.setData().forPath(SHARDING_CHANG_PATH,
                    config.getBytes());
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    public void subscribeDataChanges() {
        boolean nodeOk = ZKClient.getSingleton().isNodeExist(SHARDING_CHANG_PATH);
        if (!nodeOk) {
            ZKClient.getSingleton().createNode(SHARDING_CHANG_PATH, DEFAULT_SHARDING_RULE_CONFIG);
        }
        //注册监听事件

        CuratorFramework client = ZKClient.getSingleton().getClient();

        NodeCache nodeCache =
                new NodeCache(client, SHARDING_CHANG_PATH, false);


        NodeCacheListener listener = new NodeCacheListener() {
            @Override
            public void nodeChanged() {
                try {
                    ChildData childData = nodeCache.getCurrentData();

                    String remoteVal = new String(childData.getData(), "Utf-8");

                    String localVal = shardingRuleConfig.getConfig();


                    if (localVal.equals(remoteVal)) {
                        log.info("sharding 配置状态 localVal.equals(remoteVal) {}, {}", localVal, remoteVal);

                    } else {
                        log.info("sharding 配置状态准备改变 starting....., path={}", remoteVal);
                        shardingRuleConfig.changeConfig(remoteVal);
                    }

                } catch (UnsupportedEncodingException e) {
                    e.printStackTrace();
                }
            }
        };
        //启动节点的事件监听
        nodeCache.getListenable().addListener(listener);
        try {
            nodeCache.start();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }

    /**
     * 替换sharding里的分表规则ActualDataNodes的值
     */
    public void replaceActualDataNodes(String oldRule, String newRule) {
        // 获取已有的配置
        String rules = LocalRegistryCenter.values
                .get("/orchestration-sharding-data-source/config/schema/logic_db/rule");
        // 修改规则
        String rule = rules.replace(oldRule, newRule);
        LocalRegistryCenter.listeners.get("/orchestration-sharding-data-source/config/schema")
                .onChange(new DataChangedEvent(
                        "/orchestration-sharding-data-source/config/schema/logic_db/rule",
                        rule, DataChangedEvent.ChangedType.UPDATED));
        LocalRegistryCenter.values.put("/orchestration-sharding-data-source/config/schema/logic_db/rule", rule);
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        SpringContextUtil.setContext(applicationContext);

        delaySubscribe();
    }

    private void delaySubscribe() {
        ThreadUtil.delayRun(() -> {

            if (ZKClient.getSingleton().isReady()) {
                subscribeDataChanges();
            } else {
                //如果zk没有连接成功，这块做递归
                delaySubscribe();
            }
        }, ZKClient.DEPEND_DELAY, TimeUnit.SECONDS);
    }

    public boolean checkConfig(String config) {
        return shardingRuleConfig.checkConfig(config);
    }
}
